package pub.dsb.example.rx.operators;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

/**
 * <p>
 *
 * </p>
 *
 * @author Yi
 * @version V1.0.0
 * @date 2020/12/24 16:01
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2020/12/24 16:01
 * @modify reason: {方法名}:{原因}
 * ...
 * @since V1.0.0
 */
public class CreateTest {
    /**
     * create
     *
     * @param args
     */
    public static void main(String[] args) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
                if (!emitter.isDisposed()) {
                    int index = 5;
                    while (index-- > 0) {
                        emitter.onNext(index);
                        if (index == 3) {
                            emitter.onError(new IllegalAccessException("ILLEGAL NUMBER!"));
                            return;
                        }
                    }
                    emitter.onComplete();
                }
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer v) {
                System.out.println("onNext:" + v);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println(e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }
}
